package com.smart.tool.future;

import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @Author xiaoqiang
 * @Version CompletableFutureUtils.java, v 0.1 2025年09月13日 00:04 xiaoqiang
 * @Description: TODO
 */
@Slf4j
public final class CompletableFutureUtils {

    // 默认IO线程池（可根据业务调整）
    private static final ExecutorService DEFAULT_IO_POOL = new ThreadPoolExecutor(
            16, 32, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(10000),
            new ThreadFactory() {
                private final AtomicInteger counter = new AtomicInteger(1);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "default-io-pool-" + counter.getAndIncrement());
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    /**
     * 将 List<CompletableFuture<T>> 转换为 CompletableFuture<List<T>>
     * 用于批量任务结果聚合
     */
    public static <T> CompletableFuture<List<T>> sequence(Collection<CompletableFuture<T>> futures) {
        if (futures.isEmpty()) {
            return CompletableFuture.completedFuture(Collections.emptyList());
        }

        // 等待所有任务完成
        CompletableFuture<Void> allDone = CompletableFuture.allOf(
                futures.toArray(new CompletableFuture[0])
        );

        // 聚合结果（过滤null值）
        return allDone.thenApply(v -> futures.stream()
                .map(future -> {
                    try {
                        return future.join();  // join() 不抛受检异常，更方便
                    } catch (CompletionException e) {
                        log.error("任务执行失败", e.getCause());
                        return null;  // 失败任务返回null，后续过滤
                    }
                })
                .filter(Objects::nonNull)  // 过滤失败的任务结果
                .collect(Collectors.toList())
        );
    }

    /**
     * 并行处理集合元素，带超时和异常降级
     *
     * @param source    源集合
     * @param mapper    处理函数（将T转换为R）
     * @param timeoutMs 超时时间（毫秒）
     * @param fallback  异常或超时的兜底值
     * @return 处理后的结果集合
     */
    public static <T, R> CompletableFuture<List<R>> parallelProcess(
            Collection<T> source,
            Function<T, R> mapper,
            long timeoutMs,
            R fallback) {
        return parallelProcess(source, mapper, DEFAULT_IO_POOL, timeoutMs, fallback);
    }

    /**
     * 并行处理集合元素（支持自定义线程池）
     */
    public static <T, R> CompletableFuture<List<R>> parallelProcess(
            Collection<T> source,
            Function<T, R> mapper,
            Executor executor,
            long timeoutMs,
            R fallback) {
        if (source.isEmpty()) {
            return CompletableFuture.completedFuture(Collections.emptyList());
        }

        // 每个元素创建一个异步任务
        List<CompletableFuture<R>> futures = source.stream()
                .map(t -> CompletableFuture.supplyAsync(() -> {
                            try {
                                return mapper.apply(t);  // 执行处理函数
                            } catch (Exception e) {
                                log.error("处理元素失败: {}", t, e);
                                return fallback;  // 单个任务异常返回兜底值
                            }
                        }, executor)
                        // 单个任务超时兜底
                        .completeOnTimeout(fallback, timeoutMs, TimeUnit.MILLISECONDS))
                .collect(Collectors.toList());

        // 聚合所有结果
        return sequence(futures);
    }

    /**
     * 取多个任务中第一个成功的结果
     * 用于多数据源查询、多渠道调用等场景
     */
    public static <T> CompletableFuture<T> firstSuccess(List<CompletableFuture<T>> futures) {
        if (futures.isEmpty()) {
            return CompletableFuture.failedFuture(new IllegalArgumentException("任务列表为空"));
        }

        CompletableFuture<T> result = new CompletableFuture<>();

        // 每个任务完成时尝试设置结果（仅第一个成功的生效）
        futures.forEach(future -> future.whenComplete((res, ex) -> {
            if (ex == null && !result.isDone()) {  // 无异常且结果未设置
                result.complete(res);
            }
        }));

        // 所有任务失败时，汇总异常
        CompletableFuture<Void> allFailed = CompletableFuture.allOf(
                futures.stream()
                        .map(f -> f.exceptionally(ex -> null))  // 异常任务返回null
                        .toArray(CompletableFuture[]::new)
        );

        allFailed.whenComplete((v, ex) -> {
            if (!result.isDone()) {  // 所有任务均失败
                List<Throwable> causes = futures.stream()
                        .map(f -> {
                            try {
                                f.join();
                                return null;
                            } catch (CompletionException e) {
                                return e.getCause();
                            }
                        })
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList());
                result.completeExceptionally(
                        //new RuntimeException("所有任务均失败")
                        new AggregateException("所有任务均失败", causes)
                );
            }
        });

        return result;
    }

    /**
     * 带超时的阻塞获取结果，包装异常为RuntimeException
     * 避免调用方处理受检异常
     */
    public static <T> T getWithTimeout(CompletableFuture<T> future, long timeout, TimeUnit unit) {
        try {
            return future.get(timeout, unit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();  // 恢复中断状态
            throw new RuntimeException("任务被中断", e);
        } catch (ExecutionException e) {
            throw new RuntimeException("任务执行失败", e.getCause());
        } catch (TimeoutException e) {
            throw new RuntimeException("任务超时", e);
        }
    }

    /**
     * 关闭线程池（优雅停机）
     */
    public static void shutdownExecutor(ExecutorService executor) {
        if (executor == null || executor.isShutdown()) {
            return;
        }
        executor.shutdown();  // 禁止新任务提交
        try {
            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {  // 等待5秒
                executor.shutdownNow();  // 强制关闭
                if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                    log.error("线程池 {} 未能正常关闭", executor);
                }
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }
}
